/*
 * Copyright 2017 StreamSets Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.streamsets.pipeline.stage.origin.ipctokafka;

import com.streamsets.pipeline.api.Config;
import com.streamsets.pipeline.api.StageException;
import com.streamsets.pipeline.api.StageUpgrader;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.pipeline.stage.util.tls.TlsConfigBeanUpgradeUtil;

import java.util.ArrayList;
import java.util.List;

public class SdcIpcToKafkaUpgrader implements StageUpgrader {
  @Override
  public List<Config> upgrade(
      String library, String stageName, String stageInstance, int fromVersion, int toVersion, List<Config> configs
  ) throws StageException {
    switch(fromVersion) {
      case 1:
        upgradeV1ToV2(configs);
        if (toVersion == 2) {
          break;
        }
        //fall through
      case 2:
        upgradeV2ToV3(configs);
        if (toVersion == 3) {
          break;
        }
        //fall through
      case 3:
        upgradeV3ToV4(configs);
        break;
      default:
        throw new IllegalStateException(Utils.format("Unexpected fromVersion {}", fromVersion));
    }
    return configs;
  }

  private void upgradeV1ToV2(List<Config> configs) {
    List<Config> configsToRemove = new ArrayList<>();
    List<Config> configsToAdd = new ArrayList<>();

    // Step 1: Rename kafkaConfigBean to conf
    // Step 2: Move kafkaConfigBean.kafkaConfig.* to conf.
    for (Config config : configs) {
      if (config.getName().startsWith("kafkaConfigBean")) {
        configsToRemove.add(config);
        configsToAdd.add(new Config(
            config.getName().replace("kafkaConfigBean", "conf").replace("kafkaConfig.", ""),
            config.getValue()
        ));
      }
    }

    configs.removeAll(configsToRemove);
    configs.addAll(configsToAdd);

    // New configs are already initialized

    // TODO complete in subsequent patch
  }

  private void upgradeV2ToV3(List<Config> configs) {
    List<Config> newConfigs = new ArrayList<>();

    for (Config config : configs) {
      if (config.getName().equals("configs.maxRpcRequestSize")) {
        int sizeMB = (int) config.getValue();
        newConfigs.add(new Config(config.getName(), sizeMB * 1000));
      } else {
        newConfigs.add(config);
      }
    }
    configs.clear();
    configs.addAll(newConfigs);
  }

  private void upgradeV3ToV4(List<Config> configs) {
    TlsConfigBeanUpgradeUtil.upgradeRawKeyStoreConfigsToTlsConfigBean(
        configs,
        "configs.",
        "keyStoreFile",
        "keyStorePassword",
        "sslEnabled",
        "tlsEnabled"
    );
  }
}
